package com.izhengyin.gateway.tracing;
import com.alibaba.fastjson.JSON;
import com.izhengyin.gateway.base.ApiManageService;
import com.izhengyin.gateway.base.StatisticsService;
import com.izhengyin.gateway.base.constant.Attribute;
import com.izhengyin.gateway.base.pojo.Api;
import com.izhengyin.gateway.base.utils.HttpUtils;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.gateway.filter.GatewayFilterChain;
import org.springframework.cloud.gateway.filter.GlobalFilter;
import org.springframework.core.Ordered;
import org.springframework.http.HttpHeaders;
import org.springframework.http.server.reactive.ServerHttpRequest;
import org.springframework.lang.NonNull;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.util.StringUtils;
import org.springframework.web.server.ServerWebExchange;
import reactor.core.publisher.Mono;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author zhengyin zhengyinit@outlook.com
 * Created on 2020-09-18 16:47
 */
@Slf4j
public class GatewayTracingFilter implements GlobalFilter, Ordered {
    private final TracingConfigUtils tracingConfigUtils;
    private final ApiManageService apiManageService;
    private final StatisticsService statisticsService;
    private final Map<String,Integer> sampleCounter = new ConcurrentHashMap<>(2048);
    private final Random random = new Random();
    private final TelemetryFactory telemetryFactory;

    GatewayTracingFilter(
            TelemetryFactory telemetryFactory,
            ApiManageService apiManageService,
            StatisticsService statisticsService,
            TracingConfigUtils tracingConfigUtils
    ){
        this.telemetryFactory = telemetryFactory;
        this.apiManageService = apiManageService;
        this.statisticsService = statisticsService;
        this.tracingConfigUtils = tracingConfigUtils;

    }
    @Override
    public int getOrder() {
        return 100;
    }
    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        Api api = exchange.getAttribute(Attribute.API);
        if(Objects.isNull(api)){
            return chain.filter(exchange);
        }
        Span span;
        boolean sampled = isSampled(api);
        //是否开启采样
        if (sampled){
            span = telemetryFactory.createSpan(api);
            span.setAttribute("url", HttpUtils.getRequestFullUrl(exchange.getRequest()));
            span.setAttribute("userAgent",HttpUtils.getUserAgent(exchange.getRequest()));
            span.setAttribute("clientIp",Optional.ofNullable(exchange.getAttributes().get(Attribute.CLIENT_IP)).map(String::valueOf).orElse(""));
        }else{
            span = telemetryFactory.createNoSampledSpan();
        }
        //根据OpenTelemetry , W3c的传播协议修改请求头
        ServerHttpRequest serverHttpRequest = mutateForW3cPropagate(exchange,span);
        //记录TraceId，到请求上下文
        exchange.getAttributes().put(Attribute.TRACE_ID,span.getSpanContext().getTraceId());
        exchange.getAttributes().put(Attribute.SPAN_ID,span.getSpanContext().getSpanId());
        exchange.getAttributes().put(Attribute.SAMPLED,sampled);
        return chain.filter(exchange.mutate().request(serverHttpRequest).build())
                .doOnError(throwable -> span.setStatus(StatusCode.ERROR,throwable.getMessage()))
                .doOnSuccess(succ -> span.setStatus(StatusCode.OK))
                .doFinally(signalType -> span.end());
    }

    /**
     * 适配W3c传播链路信息
     * @param exchange
     * @param span
     * @return
     */
    private ServerHttpRequest mutateForW3cPropagate(ServerWebExchange exchange , Span span){
        return exchange.getRequest().mutate()
                .headers(headers -> W3CTraceContextPropagator.getInstance().inject(telemetryFactory.createW3CTraceContext(span), headers,HttpHeaders::add))
                .build();
    }

    /**
     * 是否开启采样
     * @param api
     * @return
     */
    private boolean isSampled(@NonNull Api api){
        //默认采样机制(每分钟最多采样1次，如果出现错误，每分钟最多采样10次)，注意是最终的采样次数需要乘以网关的节点数
        boolean sampled = Optional.of(api)
                .map(Api::getMethod)
                .map(method -> {
                    int i = this.sampleCounter.getOrDefault(method,0);
                    if(i > 0){
                        this.sampleCounter.put(method,--i);
                        return true;
                    }
                    return false;
                })
                .orElse(false);

        if(log.isDebugEnabled()){
            log.info("Api  {} 默认采样 sampled = {} ",api.getMethod(), sampled);
        }

        //根据配置的采样率进行配置
        if(!sampled){
            sampled = Optional.of(api)
                    .map(Api::getMethod)
                    .map(tracingConfigUtils::findLast)
                    .map(probability ->  probability > 0 && (int)(probability * 1000) > random.nextInt(1000))
                    .orElse(false);
            if(log.isDebugEnabled()){
                log.debug("Api {} 概率采样 sampled = {} ",api.getMethod(),sampled);
            }
        }
        return sampled;
    }


    /**
     * 默认采样控制
     *   1. 默认每分钟采样一次
     *   2. 当有错误或者平均响应时间大于1500MS采样10次
     */
    @Scheduled(initialDelay = 6000, fixedDelay = 6000)
    public void defaultSampleController(){
        this.apiManageService.getApis()
                .forEach(api -> {
                    final String name = Optional.ofNullable(api.getName()).filter(v -> !StringUtils.isEmpty(v)).orElse(api.getMethod());
                    //通过统计指标，获取上一分钟接口的的错误数与平均响应时间
                    int maxSampleNum = Optional.ofNullable(statisticsService.getLastMinuteMetrics(name))
                            .filter(summary -> {
                                boolean hasError = summary.getErrNum() > 0;
                                boolean avgTimeTooLong = summary.getAvgResTimeMs() > 1500;
                                return hasError || avgTimeTooLong;
                            })
                            .map(b -> 10)
                            .orElse(1);
                    this.sampleCounter.put(name,maxSampleNum);
                });
        log.info("sampleCounter config ， {} ", JSON.toJSONString(this.sampleCounter));
    }
}